package site.anxuan.seckill.listener;

import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import site.anxuan.seckill.entity.ItemStockLog;
import site.anxuan.seckill.service.ItemStockLogService;
import site.anxuan.seckill.service.OrderService;

@Component
@AllArgsConstructor
@RocketMQTransactionListener
public class LocalTransactionalListener implements RocketMQLocalTransactionListener {

    private final Logger logger = LoggerFactory.getLogger(LocalTransactionalListener.class);

    private final OrderService orderService;

    private final ItemStockLogService itemStockLogService;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        // 执行本地事务，判断 COMMIT 提交、ROLLBACK 回滚或者 UNKNOWN 稍后重试
        try {
            String tag = (String) message.getHeaders().get("rocketmq_TAGS");
            if (tag == null) {
                return RocketMQLocalTransactionState.COMMIT;
            }
            if (tag.equals("decrease_stock")) {
                JSONObject jsonObject = (JSONObject) arg;
                Integer userId = (Integer) jsonObject.get("userId");
                Integer itemId = (Integer) jsonObject.get("itemId");
                Integer amount = (Integer) jsonObject.get("amount");
                Integer promotionId = (Integer) jsonObject.get("promotionId");
                String itemStockLogId = (String) jsonObject.get("itemStockLogId");
                return orderService.createOrder(userId, itemId, amount, promotionId, itemStockLogId);
            } else {
                return RocketMQLocalTransactionState.UNKNOWN;
            }
        } catch (Exception e) {
            logger.error("执行本地事务出现异常.");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    @Transactional
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        // 回查本地事务执行结果
        String payload = (String) message.getPayload();
        JSONObject jsonObject = JSONObject.parseObject(payload);
        String itemStockLogId = (String) jsonObject.get("itemStockLogId");

        ItemStockLog itemStockLog = itemStockLogService.getById(itemStockLogId);
        if (itemStockLog.getStatus().equals(ItemStockLog.State.SUCCESS)) {
            return RocketMQLocalTransactionState.COMMIT;
        } else if (itemStockLog.getStatus().equals(ItemStockLog.State.FAILURE)) {
            return RocketMQLocalTransactionState.ROLLBACK;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

}
